package com.oops.consumer;

import com.oops.model.TOrder;
import com.oops.service.GoodsService;
import com.oops.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * TODO: 订单业务消费端
 *  mq默认采用异步多线程消费，无法保证消费顺序；如果要实现顺序消费，需要设置同步消费模式。
 *
 * @author wuyijie
 * @date 2022/4/16 19:18
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = "order_biz", consumerGroup = "group_2",consumeMode = ConsumeMode.ORDERLY)
public class OrderBizListener implements RocketMQListener<Long> {

    private final OrderService orderService;
    private final GoodsService goodsService;
    private final RocketMQTemplate rocketMqTemplate;

    private final ReentrantLock lock = new ReentrantLock(true);

    public OrderBizListener(OrderService orderService,GoodsService goodsService, RocketMQTemplate rocketMqTemplate) {
        this.orderService = orderService;
        this.goodsService = goodsService;
        this.rocketMqTemplate = rocketMqTemplate;
    }

    /**
     * 定时消息（延迟队列）级别
     * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    @Override
    public void onMessage(Long goodsId) {
        // 添加订单
        final TOrder order = orderService.add();

        // 加锁，扣件库存
        lock.lock();
        try {
            // 扣减库存
            goodsService.decreaseGoodsInventory(goodsId);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

        HashMap<String, Object> map = new HashMap<>();
        map.put("orderId", order.getId());
        map.put("goodsId", goodsId);

        // 发送延时消息 订单失效
        rocketMqTemplate.asyncSend("order_delay", MessageBuilder.withPayload(map).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("发送延时消息成功：{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("发送延时小时失败：", throwable);
            }
        }, 3000, 6);
    }


}
